package com.bangwz.aio;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.CountDownLatch;

public class AsyncServerHandler implements Runnable {

	public CountDownLatch latch;

	public AsynchronousServerSocketChannel channel;

	public AsyncServerHandler(int port) throws IOException {
		// 服务端通道
		channel = AsynchronousServerSocketChannel.open();
		channel.bind(new InetSocketAddress(port));
		System.err.println("server is started.");
	}

	@Override
	public void run() {
		// CountDownLatch初始化
		// 它的作用：在完成一组正在执行的操作之前，允许当前的现场一直阻塞
		// 此处，让现场在此阻塞，防止服务端执行完成后退出
		// 也可以使用while(true)+sleep
		// 生成环境就不需要担心这个问题，以为服务端是不会退出的
		latch = new CountDownLatch(1);
		// accept 处理器
		channel.accept(this, new AcceptHandler());
		try {
			latch.await();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

}

class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, AsyncServerHandler> {

	@Override
	public void completed(AsynchronousSocketChannel channel, AsyncServerHandler serverHandler) {
		Server.clientCount++;
		System.err.println("当前客户端数：" + Server.clientCount);
		serverHandler.channel.accept(serverHandler, this);

		// 创建buffer
		ByteBuffer byteBuffer = ByteBuffer.allocate(1024);

		channel.read(byteBuffer, byteBuffer, new ReadHandler(channel));
	}

	@Override
	public void failed(Throwable exc, AsyncServerHandler serverHandler) {
		exc.printStackTrace();
		serverHandler.latch.countDown();
	}
}

class ReadHandler implements CompletionHandler<Integer, ByteBuffer> {

	private AsynchronousSocketChannel channel;

	public ReadHandler(AsynchronousSocketChannel channel) {
		this.channel = channel;
	}

	@Override
	public void completed(Integer result, ByteBuffer byteBuffer) {
		byteBuffer.flip();

		byte[] msg = new byte[byteBuffer.remaining()];
		byteBuffer.get(msg);
		try {
			System.err.println("服务器收到消息：" + new String(msg, "UTF-8"));
			// 向客户端发送消息
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		}
		// 向客户端写消息
		byte[] bytes = "服务器收到消息.".getBytes();
		ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
		writeBuffer.put(bytes);
		writeBuffer.flip();
		channel.write(writeBuffer, writeBuffer, new WriteHandler(channel));
	}

	@Override
	public void failed(Throwable exc, ByteBuffer byteBuffer) {
		try {
			this.channel.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
}

class WriteHandler implements CompletionHandler<Integer, ByteBuffer> {

	private AsynchronousSocketChannel channel;

	public WriteHandler(AsynchronousSocketChannel channel) {
		this.channel = channel;
	}

	@Override
	public void completed(Integer result, ByteBuffer buffer) {
		if (buffer.hasRemaining()) {
			channel.write(buffer, buffer, this);
		} else {
			ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
			channel.read(byteBuffer, byteBuffer, new ReadHandler(channel));
		}
	}

	@Override
	public void failed(Throwable exc, ByteBuffer attachment) {
		try {
			this.channel.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

}
